#include "EventLoop.h"
#include "Logger.h"
#include "EPollPoller.h"
#include "Channel.h"
#include "TaskQueue.h"

#include <sys/eventfd.h>
#include <unistd.h>
#include <sstream>
#include <iomanip>

// 防止一个线程创建多个 eventloop
__thread EventLoop* t_loopInThisThread = 0;

TaskQueue g_taskQueue;

const int kNew = -1; // Channel成员的 index 初始化就是-1
// Channel 已添加到 Poller 中
const int kAdded = 1;

const int kDeleted = 2;

void PushToTaskQueue(Channel* channel)
{
  g_taskQueue.push(channel);
}

Channel* PopFromTaskQueue() 
{
  return g_taskQueue.pop();
}

const int kPollTimeMs = 10000; // 定义默认的 poller 接口的超时时间
// 用来唤醒 subloop 的 wakeupfd
int createEventfd()
{
  int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
  if (evtfd < 0) {
    LOG_FATAL("Failed in eventfd"); 
  }
  return evtfd;
}



EventLoop::EventLoop()
  : m_looping(false)
  , m_quit(false)
  , m_callingPendingFunctor(false)
  , m_threadId(CurrentThread::tid())
  , m_poller(Poller::newDefaultPoller(this))
  , m_wakeupFd(createEventfd())
  , m_wakeupChannel(new Channel(this, m_wakeupFd))
  , m_currentActiveChannel(nullptr)
  , m_pendingTasksCount(0)
{
  LOG_INFO("EventLoop created %p in thread %d", this, m_threadId);
  if (t_loopInThisThread) {
    LOG_FATAL("Another EventLoop is already running in this thread %d", m_threadId);
  } else {
    t_loopInThisThread = this;
  }
  
  // 设置 wakeup fd 的事件类型
  m_wakeupChannel->setReadCallback(std::bind(&EventLoop::handleRead, this));
  m_wakeupChannel->enableReading();
  LOG_INFO("EventLoop created end in thread %d", m_threadId);
}
EventLoop::~EventLoop() 
{
  m_wakeupChannel->disableAll();
  m_wakeupChannel->remove();
  close(m_wakeupFd);
  t_loopInThisThread = nullptr;
}

void EventLoop::handleRead()
{
  uint64_t one = 1;
  ssize_t n = read(m_wakeupFd, &one, sizeof(one));
  if (n != sizeof(one)) {
    LOG_ERROR("EventLoop::handleRead() reads %lu", n);
  }
}

// 这个接口肯定不可能是别的线程执行了
void EventLoop::loop()
{
  m_looping = true;
  m_quit = false;   
  LOG_INFO("EventLoop %p start looping", this);
  while (!m_quit) {
    m_activeChannels.clear();
    // epoll_wait
    m_pollReturnTime = m_poller->poll(kPollTimeMs, &m_activeChannels);
    for (const auto& channel : m_activeChannels) {
      m_currentActiveChannel = channel;
      // 通知 channel 处理相应的事件
      channel->handleEvent(m_pollReturnTime);
      m_currentActiveChannel = nullptr;
    }
    m_currentActiveChannel = nullptr;
    // 执行当前 eventloop 事件循环需要处理的回调操作
    doPendingFunctors();
  }
  LOG_INFO("EventLoop %p stop looping", this);
  m_looping = false;
}

void EventLoop::quit()
{
  if (m_quit) return;
  m_quit = true;
  m_listenCnt = 0;
  if (!isInLoopThread()) {
    wakeup();
  }
}

void EventLoop::runInLoop(Functor cb)
{
  if (isInLoopThread()) {
    cb();
  } else { // 在非 loop 线程中执行 cb，就需要唤醒 loop 所在线程执行 cb
    queueInLoop(std::move(cb));
  }
}

void EventLoop::queueInLoop(Functor cb)
{
  {
    std::unique_lock<std::mutex> lk(m_mutex);
    m_pendingFunctor.emplace_back(cb);
    m_pendingTasksCount.fetch_add(1, std::memory_order_relaxed);
  }
  // 如果不在当前线程中，就唤醒
  if (!isInLoopThread() || m_callingPendingFunctor) {
    wakeup();
  }
}

void EventLoop::wakeup()
{
  uint64_t one = 1;
  ssize_t n = write(m_wakeupFd, &one, sizeof(one));
  if (n != sizeof(one)) {
    LOG_ERROR("EventLoop::wakeup() writes %lu", n);
  }
}
void EventLoop::updateChannel(Channel* channel)
{
  m_poller->updateChannel(channel);
  int index = channel->index();
  if (index == kNew || index == kDeleted) {
    m_listenCnt++;
  } else if (!channel->isNoneEvent()) {
    m_listenCnt++;
  }
}
void EventLoop::removeChannel(Channel* channel)
{
  m_poller->removeChannel(channel);
  m_listenCnt--;
}
bool EventLoop::hasChannel(Channel* channel)
{
  return m_poller->hasChannel(channel);
}

void EventLoop::doPendingFunctors()
{
  std::vector<Functor> functors;
  m_callingPendingFunctor = true;
  size_t taskNum = 0;
  {
    std::unique_lock<std::mutex> lk(m_mutex);
    functors.swap(m_pendingFunctor);
    taskNum = functors.size();
  }
  m_pendingTasksCount.fetch_sub(taskNum, std::memory_order_relaxed);
  for (const auto& cb : functors) {
    cb(); // 执行当前 loop 需要执行的回调操作
  }
  m_callingPendingFunctor = false;
}

std::string EventLoop::getHashCode()
{
  const std::size_t FNV_OFFSET_BASIS = 14695981039346656037ull;
  const std::size_t FNV_PRIME = 1099511628211ull;

  std::size_t hash = FNV_OFFSET_BASIS;
  std::string value = std::to_string(m_threadId);
  for (char c : value) {
    hash = (hash * FNV_PRIME) ^ static_cast<std::size_t>(c);
  }
  auto toHexString = [](std::size_t hash) -> std::string {
    std::stringstream ss;
    ss << std::hex << std::setw(16) << std::setfill('0') << hash;
    return ss.str();
  };

  return toHexString(hash);
}